package school;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;

import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;


public class StudentCourse {
    Connection conn = null;
    public static TableName tName = TableName.valueOf("t_student_course");
    public static TableName tStudent = TableName.valueOf("t_student");
    public static TableName tCourse = TableName.valueOf("t_course");
    @Before
    public void getConn() throws IOException {

        //构建一个连接对象
        Configuration conf = HBaseConfiguration.create();  //会自动加载hbase-site.xml
        conf.set("hbase.zookeeper.quorum", "192.168.88.5:2181,192.168.88.6:2181,192.168.88.7:2181");
        conn = ConnectionFactory.createConnection(conf);
    }
    @Test
    public void createTable() throws IOException {
        //从连接中构造一个DDL操作器
        Admin admin = conn.getAdmin();
        //创建一个表定义描述对象
        HTableDescriptor ht = new HTableDescriptor(tName);
        //创建列族定义描述对象
        HColumnDescriptor hc = new HColumnDescriptor("cf1".getBytes());
        //设置该列族中存储数据的最大版本数，默认是1
        hc.setMaxVersions(5);
        //启用块缓存
        hc.setBlockCacheEnabled(true);
        //设置块的大小
        hc.setBlocksize(180000);
        //将列族定义信息对象放入表定义对象中
        ht.addFamily(hc);

        //用DDL操作器对象：admin来建表
        admin.createTable(ht);
        //关闭连接
        admin.close();
        conn.close();
        System.out.println("表创建完成");
    }


    @Test
    public void insert() throws IOException {
        Table table = conn.getTable(tName);
        List<Put> list = new ArrayList<Put>();
        //学生1
        String stu1 = "student2019061413515518788";
        //学生2
        String stu2 = "student2019061413515521362";
        //学生3
        String stu3 = "student2019061413515526765";
        //课程 course2019061411331911964 course2019061411331917934 course2019061411331924464
        Put put = new Put("student2019061413515518788_course2019061411331911964".getBytes());
        put.addColumn("cf1".getBytes(),"student".getBytes(),stu1.getBytes());
        Put put1 = new Put("student2019061413515518788_course2019061411331917934".getBytes());
        put1.addColumn("cf1".getBytes(),"student".getBytes(),stu1.getBytes());
        Put put2 = new Put("student2019061413515518788_course2019061411331924464".getBytes());
        put2.addColumn("cf1".getBytes(),"student".getBytes(),stu1.getBytes());
        Put put3 = new Put("student2019061413515521362_course2019061411331911964".getBytes());
        put3.addColumn("cf1".getBytes(),"student".getBytes(),stu2.getBytes());
        Put put4 = new Put("student2019061413515521362_course2019061411331917934".getBytes());
        put4.addColumn("cf1".getBytes(),"student".getBytes(),stu2.getBytes());
        Put put5 = new Put("student2019061413515526765_course2019061411331911964".getBytes());
        put5.addColumn("cf1".getBytes(),"student".getBytes(),stu2.getBytes());
        list.add(put);
        list.add(put1);
        list.add(put2);
        list.add(put3);
        list.add(put4);
        list.add(put5);
        table.put(list);
        System.out.println("数据插入完成");
    }


    /**
     *   查询某个学生所选的所有课程列表 学生1
     * @throws IOException
     */
    @Test
    public void findByCourse() throws IOException {
        //根据表名获取表
        Table table = conn.getTable(tName);
        //定义一个学生ID
        String stuId = "student2019061413515518788";
        //定义一个list集合
        List<String> list =new ArrayList<String>();
        //定义一个扫描器
        Scan scan = new Scan();
        //过滤出学生id为“student2019061413515518788”的所有课程
        RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL,new RegexStringComparator(stuId+"_"));
        //扫描条件
        scan.setFilter(filter);
        //扫描结果
        ResultScanner results = table.getScanner(scan);
        //扫描结果放入迭代器
        Iterator<Result> it = results.iterator();
        while (it.hasNext()){
            Result result = it.next();
            //把结果放入byte数组
            byte[] rowKey = result.getRow();
            //把byte数组转为utf8格式放入list集合
            list.add(new String(rowKey,"utf8"));
        }
        for(String id :list){
            //根据"_"切分把下标为1的拿出来
            String courseId = id.split("_")[1];
            //根据表名获取表
            Table courseTable = conn.getTable(tCourse);
            //根据课程id查询
            Get get = new Get(courseId.getBytes());
            Result result = courseTable.get(get);
            //查询出来的结果放入byte数组
            byte[] name = result.getValue("cf1".getBytes(),"name".getBytes());
            System.out.println("课程ID:"+courseId+" 名称：" + new String(name,"utf8"));
        }
    }
    /**
     * 查询某个课程，的学生列表 课程1
     * @throws IOException
     */
    @Test
    public void findByStudent() throws IOException {
        //根据表名获取表
        Table table = conn.getTable(tName);
        //定义一个课程ID
        String couId = "course2019061411331911964";
        //定义一个list集合
        List<String> list = new ArrayList<String>();
        //定义一个扫描器
        Scan scan = new Scan();
        //过滤出课程id为“_course2019061411331911964”的所有课程
        RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL,new RegexStringComparator("_" + couId));
        //扫描条件
        scan.setFilter(filter);
        //扫描结果
        ResultScanner results = table.getScanner(scan);
        //扫描结果放入迭代器
        Iterator<Result> it = results.iterator();
        while(it.hasNext()){
            Result result = it.next();
            //把结果放入byte数组
            byte[] rowKey = result.getRow();
            //把byte数组转为utf8格式放入list集合
            list.add(new String(rowKey,"utf8"));
        }
        for(String id : list){
            //根据"_"切分把下标为0的拿出来
            String studentId = id.split("_")[0];
            //根据表名获取表
            Table studentTable = conn.getTable(tStudent);
            //根据学生id查询
            Get get = new Get(studentId.getBytes());
            Result result = studentTable.get(get);
            //查询出来的结果放入byte数组
            byte[] name = result.getValue("cf1".getBytes(),"name".getBytes());
            System.out.println("学生ID:"+studentId+" 名字：" + new String(name,"utf8"));
        }
    }

    @Test
    public void delete() throws IOException {
        String stuId = "student2019061413515518788";
        String couId = "course2019061411331911964";
        String scId = stuId + "_" +couId;
        Delete delete = new Delete(scId.getBytes());
        Table table = conn.getTable(tName);
        table.delete(delete);
        conn.close();
        System.out.println("删除成功！！！");
    }
}
